package dse.messages.multiversion;
//distributed version server: version fetching round
import java.util.*;

import dse.*;
import dse.messages.*;
import dse.messages.singleversion.*;

public class GetVNFTransaction extends Transaction 
implements ShareDownloader, VersionReader
{
	//private Node requester;
	private DataObject desiredObject;
	private LinkedList<Node> residentNodes;
	private boolean failed;
	private LinkedList<Node> sharesDownloaded;
	
	private TreeMap<Version, LinkedList<Node>> versionsDownloaded;
	private int requestShareCount;
	private int validShareCount;
	private int maxFailure;
	private int successCount;
	private int failureCount;
	private int maxTS;
	private Version maxStableVersion;
	private int sharesToBeUpdate;
	
	public GetVNFTransaction(Experiment experiment,
			int id,
			Node requester,
			DataObject desiredObject,
			double time
			)
	{
		super(id, requester, experiment, time);
		
		this.desiredObject = desiredObject;
		this.failed = false;
		this.requestShareCount = 0;
		this.validShareCount = 0;
		this.maxFailure = experiment.modules().nodeEventScheduler().getEventsPerPeriod();
		this.successCount = 0;
		this.failureCount = 0;
		this.maxTS = 0;
		this.maxStableVersion = null;
		this.sharesToBeUpdate = desiredObject.shareCount();
		if(isUpdate && isPartialUpdate)
		{
			this.sharesToBeUpdate = (int)(desiredObject.shareCount()*Math.random());
		}
	}

	@Override
	public void start(ExperimentRecorder recorder) {
		// TODO Auto-generated method stub
		residentNodes = MessageUtil.getNodesByDht(experiment.network(), requester, desiredObject, experiment.modules().assigner());		

		if(isUpdate)
		{
			recorder.dataUpdateInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			this.requestShareCount = desiredObject.shareCount();
			this.validShareCount = requestShareCount - this.maxFailure;
			
			for (int i=0; (i<requestShareCount) && (i<residentNodes.size()); i++)
				sendMaxTSRequest(recorder, residentNodes.get(i), startTime);
		}
		else
		{
			recorder.dataRequestInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			
			this.requestShareCount = desiredObject.requiredShareCount();
			this.requestShareCount = this.requestShareCount>this.desiredObject.shareCount()?this.desiredObject.shareCount():this.requestShareCount;
			this.validShareCount = desiredObject.requiredShareCount();
			for(int i=0; (i<requestShareCount)&&(i<residentNodes.size()); i++)
			{
				sendVersionRequest(recorder, residentNodes.get(i), startTime);
			}		
		}
	}
	
	@Override
	public void shareDownloaded (ExperimentRecorder recorder, Node from,
			DataObject dobj, int shareIndex, double timeLeft)
	{
	}
	
	@Override
	public void shareDownloadFailed (ExperimentRecorder recorder, DataObject dobj, 
			int shareIndex, double timeLeft)
	{		
	}
	
	@Override
	public boolean isDoneDownloadingShares ()
	{
		return (failed || finished);
	}
	
	private void SendRequestDesignatedShareRequest(ExperimentRecorder recorder, Version version, Node node, double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestDesignatedShareMessage(requester, node, desiredObject, version,
				MessagePurpose.DataObjectRequest, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	@Override
	public void shareUpdated (ExperimentRecorder recorder, Node to,
			DataObject dobj, int shareIndex, double timeLeft)
	{
		if(isUpdate) {
			this.successCount++;
			if((successCount+failureCount) >= this.sharesToBeUpdate)
			{
				if(this.successCount >= validShareCount)
					transactionSucceed(recorder, timeLeft);
				else
					transactionFailed(recorder, timeLeft);
			}
		}
	}
	
	@Override
	public void shareUpdateFailed (ExperimentRecorder recorder, DataObject dobj, 
			int shareIndex, double timeLeft)
	{
		if(isUpdate) {
			this.failureCount++;
			if((successCount + failureCount) >= this.sharesToBeUpdate)
			{
				if(this.successCount >= validShareCount)
					transactionSucceed(recorder, timeLeft);
				else
					transactionFailed(recorder, timeLeft);
			}
		}
	}
	
	@Override
	public boolean isDoneUpdatingShares ()
	{
		return (failed || finished);
	}
	
	private void sendUpdateShareRequest(ExperimentRecorder recorder, Node node, Version version, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createShareUpdateV3Message(requester, node, desiredObject, 
				version, MessagePurpose.DataObjectUpdate, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	@Override
	public void readVersion(ExperimentRecorder recorder, TreeSet<Version> versions, double timeLeft, Node from)
	{
		if(versions == null)
		{
			failureCount++;
			if(!isUpdate){
				this.requestShareCount++;
				this.requestShareCount = this.requestShareCount>this.desiredObject.shareCount()?this.desiredObject.shareCount():this.requestShareCount;
				//System.out.println(this.requestShareCount);
				sendVersionRequest(recorder, residentNodes.get(this.requestShareCount-1), timeLeft);
			}
			return;
		}
		//System.out.println("version number list : " + versions.size());
		recorder.shareHistorySize(versions.size());
		if(versionsDownloaded == null)
			versionsDownloaded = new TreeMap<Version, LinkedList<Node>>();
			
		successCount++;
		for(Version ver : versions)
		{
			if(this.versionsDownloaded.get(ver) == null){
				this.versionsDownloaded.put(ver, new LinkedList<Node>());
				this.versionsDownloaded.get(ver).add(from);
			}
			else
				this.versionsDownloaded.get(ver).add(from);
		}
		
		if(successCount+failureCount >= requestShareCount)
			processVersions(recorder, timeLeft);
	}
	
	public void readVersion(ExperimentRecorder recorder, Node from, Version version, double timeLeft)
	{
		if(failed || finished)
			return;
		if((version == null) || (this.maxStableVersion.compareTo(version) != 0))
		{
			transactionFailed(recorder, timeLeft);
			return;
		}
		if(sharesDownloaded == null)
			sharesDownloaded = new LinkedList<Node>();
		
		//successCount++;
		this.sharesDownloaded.add(from);
		
		if(this.sharesDownloaded.size() >= this.desiredObject.requiredShareCount()){
			transactionSucceed(recorder, timeLeft);
		}
	}
	
	public void readVersion(ExperimentRecorder recorder, int seq, double time){
		
		if(seq == -1)
		{
			failureCount++;
			return;
		}
		successCount++;
		if(this.maxTS < seq)
			this.maxTS = seq;
		if(successCount+failureCount >= requestShareCount){
			successCount = 0;
			failureCount = 0;
			if(this.maxTS >= requester.getSequence())
				requester.setSequence(this.maxTS);
			Version version = new Version(requester.sequence(), requester.id());
			for(int i=0; (i<this.sharesToBeUpdate) && (i<residentNodes.size()); i++)
				sendUpdateShareRequest(recorder, residentNodes.get(i), version, time);
		}
		
	}
	public void readVersion(ExperimentRecorder recorder, int seq, int index, double time){}
	public void readVersion(ExperimentRecorder recorder, double time, LinkedList<Version> versions, Version version, Node from){}
	
	private void sendVersionRequest(ExperimentRecorder recorder, Node node, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestVersionMessage(
				requester, 
				node,
				desiredObject, 
				MessagePurpose.DataObjectUpdate,
				this
				);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void sendMaxTSRequest(ExperimentRecorder recorder, Node node, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestMaxTSMessage(
				requester, 
				node,
				desiredObject, 
				MessagePurpose.DataObjectUpdate,
				this
				);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void processVersions(ExperimentRecorder recorder, double timeLeft)
	{
		if((versionsDownloaded==null) || (versionsDownloaded.size()==0))
		{
			transactionFailed(recorder, timeLeft);
			return;
		}
		if(this.versionsDownloaded.lastKey().getSequence()>requester.getSequence())
			requester.setSequence(versionsDownloaded.lastKey().getSequence());
		
		Map.Entry<Version, LinkedList<Node>> version = versionsDownloaded.pollLastEntry();
		while((versionsDownloaded.size()>0) && (version.getValue().size() < desiredObject.requiredShareCount()))
		{
			version = versionsDownloaded.pollLastEntry();
		}
		maxStableVersion = version.getKey();
		
		this.successCount = 0;
		this.failureCount = 0;
		for(Node n : version.getValue())
			SendRequestDesignatedShareRequest(recorder, maxStableVersion, n, timeLeft);
	}
	
	private void transactionSucceed(ExperimentRecorder recorder, double timeLeft)
	{
		finished = true;
		failed  = false;
		//System.out.println("Data ID : " + desiredObject.id() + "Version : " + maxVersion.getSequence() + " , " + maxVersion.getClientID());
		if(isUpdate)
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataUpdateFinished(super.id, requester, desiredObject, true, timeLeft);
		}
		else
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataRequestFinished(super.id, requester, desiredObject, true, this.residentNodes.get(this.validShareCount-1),
					false, timeLeft);
		}
	}
	
	private void transactionFailed (ExperimentRecorder recorder, double timeLeft)
	{
		recorder.debug(this, "transFailed", timeLeft);
		
		finished = true;
		failed = true;
		if(isUpdate)
			recorder.dataUpdateFinished(super.id, requester, desiredObject, false, timeLeft);
		else
			recorder.dataRequestFinished(super.id, requester, desiredObject, false, null, false, timeLeft);
	}
	public Transaction getTransaction(){
		return this;
	}
}
